use std::any::Any;
use std::fmt::Debug;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
use tokio::task::JoinHandle;
use futures::future::FutureExt;
use futures::task::Poll;
use futures::future::poll_fn;
use std::time::Duration;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::time::Instant;
use crate::actor::{ActorError, SystemMessage, PID};
use crate::dispatcher::DispatcherTrait;
use tracing::{trace, debug, info, warn, error};

#[cfg(feature = "crossbeam_mailbox")]
use crossbeam::channel::{self, Receiver, Sender};

/// A future that is always ready with () and implements Unpin
struct ReadyFuture;

impl Future for ReadyFuture {
    type Output = ();
    
    fn poll(self: Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> std::task::Poll<Self::Output> {
        std::task::Poll::Ready(())
    }
}

impl Unpin for ReadyFuture {}

/// Common mailbox trait that defines how messages are delivered to actors
pub trait Mailbox: Send + Sync + Debug {
    /// Start the mailbox processing
    fn start(&self);
    
    /// Post a user message to the mailbox
    fn post_user_message(&self, message: Box<dyn Any + Send + Sync>) -> Result<(), ActorError>;
    
    /// Post a system message to the mailbox
    fn post_system_message(&self, message: SystemMessage) -> Result<(), ActorError>;
    
    /// Process messages from the mailbox
    fn process_messages(&self);
}

/// Message invoker trait for processing messages
pub trait MessageInvoker: Send + Sync + Debug {
    /// Invoke a user message
    fn invoke_user_message(&self, message: Box<dyn Any + Send + Sync>);
    
    /// Invoke a system message
    fn invoke_system_message(&self, message: SystemMessage, pid: &PID);
    
    /// Check if the invoker is alive
    fn is_alive(&self) -> bool;
    
    /// Escalate a failure to the parent
    fn escalate_failure(&self, reason: Box<dyn Any + Send + Sync>, message: Option<Box<dyn Any + Send + Sync>>);
}

/// Default mailbox implementation
#[derive(Debug)]
pub struct DefaultMailbox {
    user_messages: Mutex<UnboundedReceiver<Box<dyn Any + Send + Sync>>>,
    user_sender: UnboundedSender<Box<dyn Any + Send + Sync>>,
    system_messages: Mutex<UnboundedReceiver<SystemMessage>>,
    system_sender: UnboundedSender<SystemMessage>,
    invoker: Arc<dyn MessageInvoker>,
    dispatcher: Arc<dyn crate::dispatcher::DispatcherTrait>,
}

impl DefaultMailbox {
    /// Create a new default mailbox
    pub fn new(invoker: Arc<dyn MessageInvoker>, dispatcher: Arc<dyn crate::dispatcher::DispatcherTrait>) -> Self {
        let (user_sender, user_receiver) = mpsc::unbounded_channel();
        let (system_sender, system_receiver) = mpsc::unbounded_channel();
        
        Self {
            user_messages: Mutex::new(user_receiver),
            user_sender,
            system_messages: Mutex::new(system_receiver),
            system_sender,
            invoker,
            dispatcher,
        }
    }
    
    /// Schedule message processing
    fn schedule(&self) {
        trace!("[DefaultMailbox] Scheduling message processing");
        let mailbox = self.clone();
        
        self.dispatcher.schedule_boxed(Box::new(move || {
            mailbox.process_messages();
            Pin::from(Box::new(ReadyFuture) as Box<dyn Future<Output = ()> + Send + Unpin>)
        }));
    }
}

impl Clone for DefaultMailbox {
    fn clone(&self) -> Self {
        trace!("[DefaultMailbox] Cloning mailbox, self ptr: {:p}", self);
        trace!("[DefaultMailbox] Original user_sender ptr: {:p}", &self.user_sender);
        
        // Create new channels
        let (user_sender, user_receiver) = mpsc::unbounded_channel();
        let (system_sender, system_receiver) = mpsc::unbounded_channel();
        
        Self {
            user_messages: Mutex::new(user_receiver),
            user_sender,
            system_messages: Mutex::new(system_receiver),
            system_sender,
            invoker: self.invoker.clone(),
            dispatcher: self.dispatcher.clone(),
        }
    }
}

impl Mailbox for DefaultMailbox {
    fn start(&self) {
        trace!("[DefaultMailbox] Starting mailbox");
        self.schedule();
    }
    
    fn post_user_message(&self, message: Box<dyn Any + Send + Sync>) -> Result<(), ActorError> {
        trace!("[DefaultMailbox] Posting user message, type_id: {:?}", message.type_id());
        trace!("[DefaultMailbox] user_sender ptr: {:p}", &self.user_sender);
        
        match self.user_sender.send(message) {
            Ok(_) => {
                trace!("[DefaultMailbox] User message sent to channel successfully");
                
                if let Ok(receiver) = self.user_messages.try_lock() {
                    trace!("[DefaultMailbox] Successfully locked user_messages for verification");
                    trace!("[DefaultMailbox] user_messages ptr: {:p}", &*receiver);
                } else {
                    trace!("[DefaultMailbox] Could not lock user_messages for verification (normal during concurrent access)");
                }
                
                self.schedule();
                Ok(())
            },
            Err(err) => {
                error!("[DefaultMailbox] Failed to send user message: {:?}", err);
                Err(ActorError::MessageDeliveryFailed("Failed to send user message".to_string()))
            },
        }
    }
    
    fn post_system_message(&self, message: SystemMessage) -> Result<(), ActorError> {
        trace!("[DefaultMailbox] Posting system message: {:?}", message);
        match self.system_sender.send(message) {
            Ok(_) => {
                trace!("[DefaultMailbox] System message sent to channel successfully");
                self.schedule();
                Ok(())
            }
            Err(err) => {
                error!("[DefaultMailbox] Error sending system message: {:?}", err);
                Err(ActorError::Unknown("Mailbox closed".to_string()))
            },
        }
    }
    
    fn process_messages(&self) {
        trace!("[DefaultMailbox] Processing messages");
        trace!("[DefaultMailbox] self ptr: {:p}", self);
        trace!("[DefaultMailbox] user_sender ptr: {:p}", &self.user_sender);
        
        let throughput = self.dispatcher.throughput();
        let mut processed = 0;
        
        // Process system messages first
        trace!("[DefaultMailbox] Processing system messages");
        if let Ok(mut sys_messages) = self.system_messages.lock() {
            while processed < throughput {
                match sys_messages.try_recv() {
                    Ok(msg) => {
                        trace!("[DefaultMailbox] Processing system message: {:?}", msg);
                        self.invoker.invoke_system_message(msg, &PID::new("system", "system"));
                        processed += 1;
                    },
                    Err(_) => {
                        trace!("[DefaultMailbox] No more system messages to process");
                        break;
                    }
                }
            }
        } else {
            warn!("[DefaultMailbox] Failed to lock system_messages");
        }
        
        // Then process user messages
        if processed < throughput {
            trace!("[DefaultMailbox] Processing user messages");
            trace!("[DefaultMailbox] Time: {:?}", std::time::SystemTime::now());
            
            if let Ok(mut user_messages) = self.user_messages.lock() {
                trace!("[DefaultMailbox] Successfully locked user_messages for processing");
                trace!("[DefaultMailbox] user_messages ptr: {:p}", &*user_messages);
                
                // 检查当前channel状态
                trace!("[DefaultMailbox] Checking user message channel");
                
                // 尝试接收消息
                while processed < throughput {
                    match user_messages.try_recv() {
                        Ok(msg) => {
                            trace!("[DefaultMailbox] Processing user message, type_id: {:?}", msg.type_id());
                            self.invoker.invoke_user_message(msg);
                            processed += 1;
                        },
                        Err(err) => {
                            // 打印错误类型
                            trace!("[DefaultMailbox] try_recv error: {:?}", err);
                            trace!("[DefaultMailbox] No more user messages to process");
                            break;
                        }
                    }
                }
            } else {
                warn!("[DefaultMailbox] Failed to lock user_messages");
            }
        }
        
        // If we processed any messages, schedule for more
        if processed > 0 {
            trace!("[DefaultMailbox] Processed {} messages, scheduling for more", processed);
            self.schedule();
        } else {
            trace!("[DefaultMailbox] No messages processed");
        }
    }
}

/// Create an unbounded mailbox
pub fn unbounded_mailbox_producer(
    invoker: Arc<dyn MessageInvoker>,
    dispatcher: Arc<dyn crate::dispatcher::DispatcherTrait>,
) -> Box<dyn Mailbox> {
    Box::new(DefaultMailbox::new(invoker, dispatcher))
}

/// Create a bounded mailbox with the specified capacity
pub fn bounded_mailbox_producer(
    invoker: Arc<dyn MessageInvoker>,
    dispatcher: Arc<dyn crate::dispatcher::DispatcherTrait>,
    capacity: usize,
) -> Box<dyn Mailbox> {
    Box::new(BoundedMailbox::new(invoker, dispatcher, capacity))
}

/// Bounded mailbox implementation
#[derive(Debug)]
pub struct BoundedMailbox {
    inner: DefaultMailbox,
}

impl BoundedMailbox {
    /// Create a new bounded mailbox
    pub fn new(
        invoker: Arc<dyn MessageInvoker>,
        dispatcher: Arc<dyn crate::dispatcher::DispatcherTrait>,
        _capacity: usize,
    ) -> Self {
        Self {
            inner: DefaultMailbox::new(invoker, dispatcher),
        }
    }
}

impl Mailbox for BoundedMailbox {
    fn start(&self) {
        self.inner.start()
    }
    
    fn post_user_message(&self, message: Box<dyn Any + Send + Sync>) -> Result<(), ActorError> {
        self.inner.post_user_message(message)
    }
    
    fn post_system_message(&self, message: SystemMessage) -> Result<(), ActorError> {
        self.inner.post_system_message(message)
    }
    
    fn process_messages(&self) {
        self.inner.process_messages()
    }
}

#[cfg(feature = "crossbeam_mailbox")]
/// Crossbeam-based mailbox implementation for high-performance scenarios
#[derive(Debug)]
pub struct CrossbeamMailbox {
    user_messages: Mutex<Receiver<Box<dyn Any + Send + Sync>>>,
    user_sender: Sender<Box<dyn Any + Send + Sync>>,
    system_messages: Mutex<Receiver<SystemMessage>>,
    system_sender: Sender<SystemMessage>,
    invoker: Arc<dyn MessageInvoker>,
    dispatcher: Arc<dyn DispatcherTrait>,
    // 是否处于运行状态
    running: std::sync::atomic::AtomicBool,
}

#[cfg(feature = "crossbeam_mailbox")]
impl CrossbeamMailbox {
    /// Create a new unbounded crossbeam mailbox
    pub fn new_unbounded(invoker: Arc<dyn MessageInvoker>, dispatcher: Arc<dyn DispatcherTrait>) -> Self {
        let (user_sender, user_receiver) = channel::unbounded();
        let (system_sender, system_receiver) = channel::unbounded();
        
        Self {
            user_messages: Mutex::new(user_receiver),
            user_sender,
            system_messages: Mutex::new(system_receiver),
            system_sender,
            invoker,
            dispatcher,
            running: std::sync::atomic::AtomicBool::new(false),
        }
    }
    
    /// Create a new bounded crossbeam mailbox with specified capacity
    pub fn new_bounded(invoker: Arc<dyn MessageInvoker>, dispatcher: Arc<dyn DispatcherTrait>, capacity: usize) -> Self {
        let (user_sender, user_receiver) = channel::bounded(capacity);
        let (system_sender, system_receiver) = channel::bounded(capacity);
        
        Self {
            user_messages: Mutex::new(user_receiver),
            user_sender,
            system_messages: Mutex::new(system_receiver),
            system_sender,
            invoker,
            dispatcher,
            running: std::sync::atomic::AtomicBool::new(false),
        }
    }
    
    /// Schedule the mailbox for processing
    fn schedule(&self) {
        if self.running.compare_exchange(
            false,
            true,
            std::sync::atomic::Ordering::SeqCst,
            std::sync::atomic::Ordering::SeqCst,
        ).is_ok() {
            let mailbox = Arc::new(self.clone());
            
            self.dispatcher.schedule_boxed(Box::new(move || {
                mailbox.process_messages();
                Pin::from(Box::new(ReadyFuture) as Box<dyn Future<Output = ()> + Send + Unpin>)
            }));
        }
    }
}

#[cfg(feature = "crossbeam_mailbox")]
impl Clone for CrossbeamMailbox {
    fn clone(&self) -> Self {
        Self {
            user_messages: Mutex::new(channel::unbounded().1),
            user_sender: self.user_sender.clone(),
            system_messages: Mutex::new(channel::unbounded().1),
            system_sender: self.system_sender.clone(),
            invoker: self.invoker.clone(),
            dispatcher: self.dispatcher.clone(),
            running: std::sync::atomic::AtomicBool::new(
                self.running.load(std::sync::atomic::Ordering::SeqCst)
            ),
        }
    }
}

#[cfg(feature = "crossbeam_mailbox")]
impl Mailbox for CrossbeamMailbox {
    fn start(&self) {
        self.schedule();
    }
    
    fn post_user_message(&self, message: Box<dyn Any + Send + Sync>) -> Result<(), ActorError> {
        match self.user_sender.send(message) {
            Ok(_) => {
                self.schedule();
                Ok(())
            },
            Err(_) => Err(ActorError::Unknown("Failed to send user message, channel closed".to_string())),
        }
    }
    
    fn post_system_message(&self, message: SystemMessage) -> Result<(), ActorError> {
        match self.system_sender.send(message) {
            Ok(_) => {
                self.schedule();
                Ok(())
            },
            Err(_) => Err(ActorError::Unknown("Failed to send system message, channel closed".to_string())),
        }
    }
    
    fn process_messages(&self) {
        // 处理系统消息
        let mut system_done = false;
        
        // 先尝试处理所有系统消息
        while !system_done {
            if let Ok(mut guard) = self.system_messages.try_lock() {
                match guard.try_recv() {
                    Ok(msg) => self.invoker.invoke_system_message(msg, &PID::new("system", "system")),
                    Err(channel::TryRecvError::Empty) => system_done = true,
                    Err(channel::TryRecvError::Disconnected) => system_done = true,
                }
            } else {
                system_done = true;
            }
        }
        
        // 处理用户消息
        let throughput = self.dispatcher.throughput();
        let mut user_messages_processed = 0;
        
        // 尝试处理用户消息，最多处理throughput条
        while user_messages_processed < throughput {
            if let Ok(mut guard) = self.user_messages.try_lock() {
                match guard.try_recv() {
                    Ok(msg) => {
                        self.invoker.invoke_user_message(msg);
                        user_messages_processed += 1;
                    },
                    Err(channel::TryRecvError::Empty) => break,
                    Err(channel::TryRecvError::Disconnected) => break,
                }
            } else {
                break;
            }
        }
        
        // 如果还有消息，重新调度
        if user_messages_processed >= throughput {
            self.schedule();
        } else {
            // 标记为不再运行
            self.running.store(false, std::sync::atomic::Ordering::SeqCst);
            
            // 双重检查，确保不会错过消息
            if let Ok(guard) = self.user_messages.try_lock() {
                if !guard.is_empty() {
                    self.schedule();
                }
            }
            
            if let Ok(guard) = self.system_messages.try_lock() {
                if !guard.is_empty() {
                    self.schedule();
                }
            }
        }
    }
}

#[cfg(feature = "crossbeam_mailbox")]
/// Producer function for creating an unbounded crossbeam mailbox
pub fn unbounded_crossbeam_mailbox_producer(invoker: Arc<dyn MessageInvoker>, dispatcher: Arc<dyn DispatcherTrait>) -> Box<dyn Mailbox> {
    Box::new(CrossbeamMailbox::new_unbounded(invoker, dispatcher))
}

#[cfg(feature = "crossbeam_mailbox")]
/// Producer function for creating a bounded crossbeam mailbox
pub fn bounded_crossbeam_mailbox_producer(invoker: Arc<dyn MessageInvoker>, dispatcher: Arc<dyn DispatcherTrait>, capacity: usize) -> Box<dyn Mailbox> {
    Box::new(CrossbeamMailbox::new_bounded(invoker, dispatcher, capacity))
}

/// Dispatcher adapter for converting dispatcher_enum::DispatcherTrait to dispatcher::DispatcherTrait
pub struct DispatcherEnumAdapter {
    inner: Arc<dyn crate::dispatcher_enum::DispatcherTrait>,
}

impl DispatcherEnumAdapter {
    /// Create a new dispatcher adapter
    pub fn new(inner: Arc<dyn crate::dispatcher_enum::DispatcherTrait>) -> Self {
        Self { inner }
    }
}

// Debug implementation for DispatcherEnumAdapter
impl std::fmt::Debug for DispatcherEnumAdapter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DispatcherEnumAdapter")
            .field("inner", &"<DispatcherTrait instance>")
            .finish()
    }
}

impl DispatcherTrait for DispatcherEnumAdapter {
    fn schedule_boxed(
        &self,
        f: Box<dyn FnOnce() -> Pin<Box<dyn Future<Output = ()> + Send + Unpin>> + Send>,
    ) -> JoinHandle<()> {
        // 使用tokio::spawn直接执行函数并返回JoinHandle
        tokio::spawn(async move {
            // 首先执行f()获取Future
            let fut = f();
            // 然后等待Future完成
            fut.await;
        })
    }
    
    fn throughput(&self) -> usize {
        self.inner.throughput()
    }
}

/// Wrap a dispatcher_enum::DispatcherTrait in a dispatcher::DispatcherTrait
pub fn wrap_dispatcher_enum(
    dispatcher: Arc<dyn crate::dispatcher_enum::DispatcherTrait>
) -> Arc<dyn DispatcherTrait> {
    Arc::new(DispatcherEnumAdapter::new(dispatcher))
} 